package com.sunzm.flink.pageview

import java.time.Duration

import com.sunzm.flink.pageview.bean.{ItemViewCount, UserBehavior}
import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.functions.{AggregateFunction, FlatMapFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

object StreamPageViewJob {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {

    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }


    //为了便于观察，并行度设置为1，实际生产环境不能设置成1
    env.setParallelism(1)
    //读取数据
    //val dataStream: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)
    //读取文件
    val dataStream: DataStream[String] = env.readTextFile("data/flink/case-demo/UserBehavior.csv")

    //创建一个Watermark生成策略对象，并提取事件时间
    val watermarkStrategy = WatermarkStrategy
      .forBoundedOutOfOrderness[UserBehavior](Duration.ofSeconds(0))
      .withTimestampAssigner(new SerializableTimestampAssigner[UserBehavior] {
        override def extractTimestamp(element: UserBehavior, recordTimestamp: Long): Long = {
          //因为数据中的时间单位是秒，所以这里需要乘以1000，变成毫秒
          element.timestamp * 1000L
        }
      })


    val mapedDataStream: DataStream[UserBehavior] = dataStream.flatMap(new FlatMapFunction[String, UserBehavior] {
      override def flatMap(line: String, out: Collector[UserBehavior]): Unit = {

        if (StringUtils.isNotBlank(line)) {
          //只输出不为空的数据

          val fields = StringUtils.splitByWholeSeparatorPreserveAllTokens(line, ",")

          if (fields.length >= 5) {
            val userId = fields(0).toLong
            val itemId = fields(1).toLong
            val categoryId = fields(2).toInt
            val behavior = fields(3)
            val timestamp = fields(4).toLong

            //把behavior转换成大写字母,可以兼容 pv, Pv, pV, PV
            if (StringUtils.equals("PV", behavior.toUpperCase.trim)) {
              //只输出 pv 的数据
              out.collect(UserBehavior(userId, itemId, categoryId, behavior, timestamp))
            }
          }

        }
      }
    })
      // 指定WaterMark生成策略，以及提取事件时间
      .assignTimestampsAndWatermarks(watermarkStrategy)

    //先使用全窗口函数计算
    val itemClountDS = mapedDataStream
      .keyBy(_.behavior)
      .window(TumblingEventTimeWindows.of(Time.hours(1)))
      .aggregate(new ItemCountAgg, new WindowCountResult)

    itemClountDS.print()

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }


  /**
   * 自定义预聚合函数类，每来一个数据就count加1
   */
  private class ItemCountAgg extends AggregateFunction[UserBehavior, Int, Int] {
    //创建一个用于保存中间聚合结果的累加器，这里进行初始化
    override def createAccumulator(): Int = 0

    //来一条数据，就应该+1
    override def add(value: UserBehavior, accumulator: Int): Int = {
      accumulator + 1
    }

    //获取结果
    override def getResult(accumulator: Int): Int = accumulator

    override def merge(a: Int, b: Int): Int = a + b
  }

  /**
   * 窗口函数
   */
  private class WindowCountResult extends WindowFunction[Int, ItemViewCount, String, TimeWindow] {
    override def apply(key: String,
                       window: TimeWindow,
                       input: Iterable[Int],
                       out: Collector[ ItemViewCount]): Unit = {
      //获取点击数量
      /*var count = 0
      input.foreach(ct => {
        count += ct
      })*/
      //因为我们做过预聚合，这里得到的数据只有一条
      val count = input.iterator.next()

      //获取窗口开始时间和结束时间
      val windowStart = window.getStart
      val windowEnd = window.getEnd

      //时间格式化
      val fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss")

      val windowStartStr = fastDateFormat.format(windowStart)
      val windowEndStr = fastDateFormat.format(windowEnd)

      //输出结果
      out.collect(ItemViewCount(windowStartStr, windowEndStr, count))
    }
  }

}
